package com.eyas.software.pattern.eventbus.dispatcher.impl;

import com.eyas.software.pattern.eventbus.dispatcher.Dispatcher;
import com.eyas.software.pattern.eventbus.event.Event;
import com.eyas.software.pattern.eventbus.register.Register;
import com.eyas.software.pattern.eventbus.subscriber.Subscriber;

import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;

public class SyncDispatcher implements Dispatcher, Closeable {
    //定义线程池
    private ExecutorService executorService = new ThreadPoolExecutor(10, 30, 300,
            TimeUnit.SECONDS, new LinkedBlockingQueue<>(30));

    @Override
    public void dispatch(Register register, Event event) {
        this.executorService.submit(new SyncDsipathcerThread(register, event));
    }

    @Override
    public void close() throws IOException {
        this.executorService.shutdown();
    }

    private static class SyncDsipathcerThread implements Runnable {
        private Register register;
        private Event event;

        SyncDsipathcerThread(Register register, Event event) {
            this.register = register;
            this.event = event;
        }

        @Override
        public void run() {
            List<Subscriber> subscribers = register.getSubscribers(event);
            subscribers.parallelStream().forEach(subscriber -> {
                subscriber.invoke(event);
            });
        }
    }
}
